package com.zyh.flink.day04;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

public class CommentBroadcastStateJob2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //低速流获取(敏感词 +|-)
        SingleOutputStreamOperator<Tuple2<String, String>> keywordsStream = environment.socketTextStream("hadoop10", 9991)
                .map(new MapFunction<String, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(String s) throws Exception {
                        String[] ss = s.split("\\s+");
                        return Tuple2.of(ss[0],ss[1]);
                    }
                });
        DataStreamSource<String> commentsStream = environment.socketTextStream("hadoop10", 9992);

        //1 广播低速流
        /*
        * 广播低速流,需要接受一个MapStateDescriptor,广播状态本质上属于MapState
        * 向广播状态中添加的键值对
        * key固定为"keyword",value为Set集合保存多个敏感词
        * */
        MapStateDescriptor<String, Set<String>> msd = new MapStateDescriptor<String, Set<String>>("msd", String.class, (Class<Set<String>>) (Class<?>) Set.class);
        BroadcastStream<Tuple2<String, String>> broadcastStream = keywordsStream.broadcast(msd);
        //2 高速流连接广播流,形成广播连接流
        BroadcastConnectedStream<String, Tuple2<String, String>> connectedStream = commentsStream.connect(broadcastStream);
        //3 基于广播连接流调用process算子
        SingleOutputStreamOperator<String> processStream = connectedStream.process(new MyBroadcastProcessFunction2(msd));

        processStream.print();

        environment.execute("BroadcastStateJob2");
    }
}

/*
* IN1 高速流的元素类型
* IN2 低速流的元素类型
* OUT 输出结果类型
* */
class MyBroadcastProcessFunction2 extends BroadcastProcessFunction<String,Tuple2<String,String>,String>{
    //状态标识符
    private MapStateDescriptor<String,Set<String>> msd;
    //构造方法
    MyBroadcastProcessFunction2(MapStateDescriptor<String,Set<String>> msd){
        this.msd = msd;
    }
    @Override
    public void processElement(String comment, ReadOnlyContext readOnlyContext, Collector<String> collector) throws Exception {
        //获取广播状态(只读)
        ReadOnlyBroadcastState<String, Set<String>> broadcastState = readOnlyContext.getBroadcastState(this.msd);
        //获取其中的数据
        Set<String> keywordSet = broadcastState.get("keyword");
        //如果不为空则遍历替换
        if (keywordSet!=null){
            for (String word : keywordSet) {
                comment = comment.replace(word,"**");
            }
        }
        //将评论输出
        collector.collect(comment);
    }

    @Override
    public void processBroadcastElement(Tuple2<String, String> value, Context context, Collector<String> collector) throws Exception {
        //得到敏感词
        String keyword = value.f0;
        //得到操作标识符 +|-
        String operator = value.f1;
        //获取广播状态
        BroadcastState<String, Set<String>> broadcastState = context.getBroadcastState(this.msd);

        //从广播状态中获取Set集合
        Set<String> keywordSet = broadcastState.get("keyword");
        //第一次获取集合必定为空,需手动创建
        if (keywordSet == null){
            keywordSet = new HashSet<>();
        }
        //如果标识符为+,则添加该敏感词
        if ("+".equals(operator)){
            keywordSet.add(keyword);
        }else if ("-".equals(operator)){//如果标识符为-,则删除该敏感词
            keywordSet.remove(keyword);
        }
        //将敏感词集合放入广播状态中
        broadcastState.put("keyword",keywordSet);
    }
}